package com.broada.apm.utils;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.plugin.deletebyquery.DeleteByQueryPlugin;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * ESClient 连接工具类
 * 
 * @author louyi Create By 2014年11月24日 下午1:41:47
 */
public class EsClientUtil {
	private static Logger log = LoggerFactory.getLogger(EsClientUtil.class);
	private static Node[] nodes = null;
	private static Client client;
	private static String ip = "127.0.0.1";
	private static String clusterName;
	private static int DEFAULT_TRANSPORT = 9300;
	private static int DEFAULT_HTTPPORT = 9200;
	static {
		loadEsYml();
	}

	private static void loadEsYml() {
//		String filePath = System.getProperty("es.config.path", "/es/config/");
//		Yaml yaml = new Yaml();
//		try {
//			File file = ApmFileUtils.getFile(filePath);
//			File[] ymls = file.listFiles(new FilenameFilter() {
//				@Override
//				public boolean accept(File dir, String name) {
//					return name.endsWith(".yml") && name.startsWith("elasticsearch");
//				}
//			});
//			if (ArrayUtils.isEmpty(ymls))
//				throw new FileNotFoundException("目录\"" + file.getAbsolutePath() + "\"下不存在elasticsearch*.yml文件.");

//			nodes = new Node[ymls.length];
//			for (int i = 0; i < ymls.length; i++) {
//				Node node = new Node();
//				InputStream input = new FileInputStream(ymls[i]);
//				Map<String, Object> map = (Map<String, Object>) yaml.load(input);
//				String clusterName = (String) map.get("cluster.name");
//				int transport_tcp_port = DEFAULT_TRANSPORT;
//				if (map.get("transport.tcp.port") != null)
//					transport_tcp_port = (int) map.get("transport.tcp.port");
//				int http_port = DEFAULT_HTTPPORT;
//				if (map.get("http.port") != null)
//					http_port = (int) map.get("http.port");
//				node.setClusterName(clusterName);
//				node.setTransPort(transport_tcp_port);
//				node.setHttpPort(http_port);
//				if (map.get("network.bind_host") != null) {
//					ip = (String)map.get("network.bind_host");
//					node.setHost(ip);
//				} else { 
//					node.setHost(ip);
//				}
//				nodes[i] = node;
//			}
//			// clusterName 暂时取第一个
//			clusterName = nodes[0].getClusterName();
//			log.info(String.format("加载es.Config.path成功,配置信息如下:\nes集群名称:%s\nes集群IP:%s", clusterName, ip));
//		} catch (FileNotFoundException e) {
//			e.printStackTrace();
//		} catch (Exception e) {
//			log.error("[es.Config.path]加载失败...", e);
//		}
		clusterName = ConfigureProperties.getString("es.cluster.name", "elas.apm");
		String esIps = ConfigureProperties.getString("es.network.bind_host", ip);
		String transport_tcp_port = ConfigureProperties.getString("es.transport.tcp.port");
		String http_port = ConfigureProperties.getString("es.http.port");
		String[] ips = null;
		if (esIps.contains(",")) {
			ips = esIps.split(",");
		} else {
			ips = new String[]{esIps};
		}
		
		nodes = new Node[ips.length];
		for (int i=0; i<ips.length; i++) {
			Node node = new Node();
			node.setClusterName(clusterName);
			node.setTransPort(StringUtils.isBlank(transport_tcp_port) ? DEFAULT_TRANSPORT : Integer.parseInt(transport_tcp_port));
			node.setHost(ips[i]);
			node.setHttpPort(StringUtils.isBlank(http_port) ? DEFAULT_HTTPPORT : Integer.parseInt(http_port));
			nodes[i] = node;
		}
		clusterName = nodes[0].getClusterName();
		log.info(String.format("加载es.Config.path成功,配置信息如下:\nes集群名称:%s\nes集群IP:%s", clusterName, ips));
	}

	public static Client getTransportClient() {
		if (client == null) {
			Settings settings = Settings.settingsBuilder().put("cluster.name", clusterName)
					.put("client.transport.sniff", false).build();
			client = TransportClient.builder().settings(settings).build();
			/**
			 * 至少添加一个InetSocketTransportAddress，否则会出现异常：
			 * org.elasticsearch.client.transport.NoNodeAvailableException: No node
			 * available。
			 */
			for (Node node : nodes) {
				try {
					((TransportClient) client).addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(node.getHost()), Integer.valueOf(node
							.getTransPort())));
				} catch (UnknownHostException e) {
					log.error("UnknownHostException");
				}
			}
		}
          
		return client;
	}
	
	
	/**
	 * 获取带删除插件的client 
	 */
	public static Client getDelteTransportClient() {
		if (client == null) {
			Settings settings = Settings.settingsBuilder().put("cluster.name", clusterName)
					.put("client.transport.sniff", false).build();
			client = TransportClient.builder().settings(settings).addPlugin(DeleteByQueryPlugin.class).build();
			/**
			 * 至少添加一个InetSocketTransportAddress，否则会出现异常：
			 * org.elasticsearch.client.transport.NoNodeAvailableException: No node
			 * available。
			 */
			for (Node node : nodes) {
				try {
					((TransportClient) client).addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(node.getHost()), Integer.valueOf(node
							.getTransPort())));
				} catch (UnknownHostException e) {
					log.error("UnknownHostException");
				}
			}
		}
          
		return client;
	}

	/**
	 * 测试链接
	 * 
	 * @return
	 * @throws Exception
	 */
	public static boolean verifyConnection() {
		List<DiscoveryNode> nodes = ((TransportClient) getTransportClient()).connectedNodes();
		if (nodes.isEmpty()) {
			log.error("No nodes available.请确认Elasticsearch运行正常!");
			return false;
		}
		return true;
	}

	/**
	 * 取任一节点 node httpPort
	 * 
	 * @return
	 */
	public static String getEsHttpUrl() {
		List<DiscoveryNode> disNodes = ((TransportClient) getTransportClient()).connectedNodes();
		if (!disNodes.isEmpty()) {
			for (DiscoveryNode disNode : disNodes) {
				InetSocketTransportAddress transportAddress = (InetSocketTransportAddress) disNode.address();
				InetSocketAddress socketAddress = transportAddress.address();
				int transPort = socketAddress.getPort();
				for (Node node : nodes) {
					if (node.getTransPort() == transPort)
						return String.format("http://%s:%s", node.getHost() == null ? ip : node.getHost(), node.getHttpPort());
				}

			}
		}
		ExceptionUtils.error(log, "No nodes available.请确认Elasticsearch运行正常!", new Exception());
		return null;
	}

}
